package com.vlin.config.tuya;

import java.util.List;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.impl.TopicMessageIdImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.vlin.api.mq.client.MessageVO;
import com.vlin.api.mq.utils.AESBase64Utils;
import com.vlin.message.handler.MessageHandler;

@Configuration
public class MessageHandlerConfig {
	
	private static final Logger logger = LoggerFactory.getLogger(MessageHandlerTask.class);
	
	
	@Bean
	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
	public MessageHandlerTask messageHandler(Integer consumerNum, Consumer<?> consumer) {
		
		return new MessageHandlerTask(consumerNum, consumer);
	}

	public static class MessageHandlerTask implements Runnable {
		
		@Autowired
		private PulsarProperty pulsarProperty;
		
		@Autowired
		private List<MessageHandler> listHandler;

		private Consumer<?> consumer;
		private Integer consumerNum;

		public MessageHandlerTask(Integer consumerNum, Consumer<?> consumer) {
			this.consumerNum = consumerNum;
			this.consumer = consumer;
		}

		@Override
		public void run() {
			do {
				try {
					Message<?> message = consumer.receive();
					Long start = System.currentTimeMillis();
					logger.info("--consumer num :{} name:{} --receive data", consumerNum,  consumer.getConsumerName());
					logger.debug("Message Received:{},seq={},time={},consumed time={},partition={}", new String(message.getData()), message.getSequenceId(),
							message.getPublishTime(), System.currentTimeMillis(), ((TopicMessageIdImpl) message.getMessageId()).getTopicPartitionName());
					String jsonMessage = new String(message.getData());
					MessageVO vo = JSON.parseObject(jsonMessage, MessageVO.class);
					String result = AESBase64Utils.decrypt(vo.getData(), pulsarProperty.getAccessKey().substring(8, 24));
					logger.info("Decrypt Message Data:{}", result);
					JSONObject body = JSON.parseObject(result);
					if(listHandler != null) {
						for(MessageHandler handler : listHandler) {
							if(handler.support(body)) {
								handler.handle(body);
							}
						}
					}
					
					Long businessCost = System.currentTimeMillis() - start;
					Long ackStart = System.currentTimeMillis();
					consumer.acknowledge(message);
					logger.debug("Cost Time: messageKey={}; business process message cost = {}ms; msg ack cost = {}ms", message.getKey(), businessCost, (System.currentTimeMillis() - ackStart));
					
				} catch (Throwable t) {
					logger.error("error:{}", t);
				}
			} while (true);
		}

		public PulsarProperty getPulsarProperty() {
			return pulsarProperty;
		}

		public void setPulsarProperty(PulsarProperty pulsarProperty) {
			this.pulsarProperty = pulsarProperty;
		}

		public List<MessageHandler> getListHandler() {
			return listHandler;
		}

		public void setListHandler(List<MessageHandler> listHandler) {
			this.listHandler = listHandler;
		}
		
	}
}
